package my.binlog;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class MySQLBinLogListener implements BinaryLogClient.EventListener {

    private static final Logger log = LoggerFactory.getLogger(MySQLBinLogListener.class);

    private final BinaryLogClient binaryLogClient;

    private final ExecutorService consumer;

    // 存放每张数据表对应的listener
    private final Map<String, List<BinLogListener>> listeners;

    private String database;
    private String tableName;

    public MySQLBinLogListener(MySQLConfig conf) {
        BinaryLogClient client = new BinaryLogClient(conf.host, conf.port, conf.username, conf.password);
        client.setConnectTimeout(10 * 60 * 60 * 1000);
//        client.setGtidSet("d108cb1a-1595-11ee-b569-00163e238eab:1");
        client.setBinlogPosition(4);
        client.setBinlogFilename("bin.000010");
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        client.setEventDeserializer(eventDeserializer);
        this.binaryLogClient = client;
        this.consumer = new ThreadPoolExecutor(1, 1,
                1, TimeUnit.HOURS,
                new SynchronousQueue<>(), new ThreadFactory() {
            private final AtomicInteger idx = new AtomicInteger(0);
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("binlog-parse-" + idx.getAndIncrement());
                return thread;
            }
        }, new ThreadPoolExecutor.CallerRunsPolicy());
        listeners = new HashMap<>();
    }

    private void execute(Event event) {
        consumer.execute(() -> {
            List<BinLogListener> listenerList = listeners.get(getTableName(database, tableName));
            if (listenerList == null) {
                log.debug("database=[{}] table=[{}] not listen", database, tableName);
                return;
            }
            BinLogItem item = new BinLogItem(database, tableName, event);
            for (BinLogListener listener : listenerList) {
                listener.onEvent(item);
            }
        });
    }

    public String getTableName(String database, String tableName) {
        return database + "*_*" + tableName;
    }

    @Override
    public void onEvent(Event event) {
        EventType eventType = event.getHeader().getEventType();

        if (eventType == EventType.TABLE_MAP) {
            TableMapEventData tableData = event.getData();
            database = tableData.getDatabase();
            tableName = tableData.getTable();
        }

        // 只处理添加删除更新三种操作
        if (EventType.isRowMutation(eventType)) {
            execute(event);
        }
    }

    public void regListener(String db, String table, BinLogListener listener) {
        listeners.computeIfAbsent(getTableName(db, table), k -> new ArrayList<>()).add(listener);
    }

    public void connect() throws IOException {
        binaryLogClient.registerEventListener(this);
        binaryLogClient.connect();
    }
}
